package com.wangwei.spring.rabbit;

import com.rabbitmq.client.Channel;
import com.wangwei.spring.rabbit.pojo.Order;
import com.wangwei.spring.rabbit.pojo.Person;
import com.wangwei.spring.rabbit.pojo.User;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 *
 *  一：
 *  SimpleMessageListenerContainer 对象可以托管到spring容器中，由spring容器进行SimpleMessageListenerContainer生命周期的管理
 *  默认情况下会在spring容器启动的时候，启动SimpleMessageListenerContainer，spring容器关闭的时候，会stop掉SimpleMessageListenerContainer
 *  也可以设置SimpleMessageListenerContainer手动启动
 *
 *  二：
 *  自动手工构造SimpleMessageListenerContainer，手工启动和停止进行消息的消费
 *  三：
 *  当同一个queue上有多个消费者的时候，只会有一个消费者收到消息
 *  四：
 *  SimpleMessageListenerContainer 支持一次监听多个queue
 *  五：
 *  SimpleMessageListenerContainer 支持运行时动态增加监听的queue
 *  六：
 *  SimpleMessageListenerContainer 支持运行时动态移除监听的queue
 *  七：
 *  setAfterReceivePostProcessors方法可以对消息进行后置处理
 *  八：
 *  setConsumerTagStrategy可以设置消费者的 Consumer tag
 *  九：
 *  setConsumerArguments 可以设置消费者的Arguments
 *  十：
 *    setConcurrentConsumers设置多少个并发消费者一起消费，并支持运行时动态修改。
 *    setMaxConcurrentConsumers设置最多的并发消费者
 *    - 动态设置并发消费者个数（setConcurrentConsumers）时，个数不能大于setMaxConcurrentConsumers所设置的值，
 *      否则会报异常：
 *      java.lang.IllegalArgumentException: 'concurrentConsumers' cannot be more than 'maxConcurrentConsumers'
 * @date 2019-10-11 20:30
 */
//@Configuration
public class SpringMQConfig {

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://wangwei:wangwei@192.168.109.128:5672");
        //CacheModel:CHANNEL-默认，可以自动声明相关组件；CONNECTION-不会自动声明
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);//true-默认，可自动声明；false-则不会自动声明
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainerWithAck(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        messageListenerContainer.setQueueNames("sms");

        /**
         * AcknowledgeMode.NONE 自动确认，等效于 autoAck=true
         *
         * AcknowledgeMode.MANUAL 手动确认，等效于 autoAck=false

         * AcknowledgeMode.AUTO 根据方法的执行情况来决定是否是确认还是拒绝（是否重新入queue）
         * 1: 如果消息成功被消费了，则自动确认
         * 2：
         *    (1)当抛出AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue=false
         *    (2)当抛出ImmediateAcknowledgeAmqpException异常，则消息会被确认
         *    (3)其他的异常，则消息会被拒绝，且requeue=true（如果此时只有一个消费者监听该队列，则有发生死循环的风险）
         *       其中，requeue=true可以通过setDefaultRequeueRejected（默认为true）去设置
         *
         *    要想了解更新详细的代码，则可以分析SimpleMessageListenerContainer.doReceiveAndExecute方法
         *
         */
//        messageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        messageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
        messageListenerContainer.setAutoStartup(true);
        /**
         * 消息消费失败，被拒绝后是否重新入队列。默认true
         */
//        messageListenerContainer.setDefaultRequeueRejected(false);
        messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                System.out.println("---------开始消息消费-------");
                System.out.println(new String(message.getBody()));
                TimeUnit.SECONDS.sleep(12);
//                if(message.getMessageProperties().getHeaders().get("error") == null){
//                    //消息确认
//                    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
//                    System.out.println("消息已确认......");
//                }else{
//                    channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
//                    System.out.println("........消息已拒绝..");
//                }

                if(StringUtils.isEmpty(message.getMessageProperties().getHeaders().get("exception"))){
                    System.out.println("-------消息自动确认------");
                }else{
                    if("1".equals(message.getMessageProperties().getHeaders().get("exception"))){
                        throw new AmqpRejectAndDontRequeueException("message consume fail!");
                    }else if("2".equals(message.getMessageProperties().getHeaders().get("exception"))){
                        throw new ImmediateAcknowledgeAmqpException("message consume fail!");
                    }else{
                        throw new RuntimeException("message consume fail!");
                    }
                }


            }
        });
        return messageListenerContainer;
    }

//    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames("sms");
        //autoStartup设置成false：不自动自动。消息不会被消费
        simpleMessageListenerContainer.setAutoStartup(true);
        //设置消息后置处理器
        simpleMessageListenerContainer.setAfterReceivePostProcessors(msg -> {
            System.out.println("------原始消息--------");
            System.out.println(msg.getMessageProperties());
            System.out.println(new String(msg.getBody()));
            msg.getMessageProperties().getHeaders().put("desc","这是一条后置处理过的消息");
            return new Message(("Message["+new String(msg.getBody())+"]").getBytes(),msg.getMessageProperties());
        });

        //设置ConsumerTag生成策略
        simpleMessageListenerContainer.setConsumerTagStrategy(queue -> {
            return "order_admin_"+queue+"_platform_"+System.currentTimeMillis();
        });

        //设置ConsumerArguments
        Map<String,Object> arguments = new HashMap<>();
        arguments.put("module","订单消息后台管理");
        arguments.put("tips","-对订单信息针对不同的消息类型进行推送-");
        simpleMessageListenerContainer.setConsumerArguments(arguments);

        //setConcurrentConsumers设置多少个并发消费者一起消费，并支持运行时动态修改。
        //setMaxConcurrentConsumers设置最多的并发消费者
//        simpleMessageListenerContainer.setConcurrentConsumers(3);
        simpleMessageListenerContainer.setMaxConcurrentConsumers(10);

        //参数只能是：接口MessageListener或ChannelAwareMessageListener的实现类
        simpleMessageListenerContainer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                System.out.println("-------------接收到队列:"+message.getMessageProperties().getConsumerQueue()+"de消息拉------------");
                System.out.println(message.getMessageProperties());
//                System.out.println(new String(message.getBody()));
                String consumerQueue = message.getMessageProperties().getConsumerQueue();
                if("sms".equals(consumerQueue)){
                    System.out.println(">>>>>>>发送短信消息: "+new String(message.getBody()));
                }else if("mail".equals(consumerQueue)){
                    System.out.println(">>>>>>>发送邮件消息: "+new String(message.getBody()));
                }else if("weixin".equals(consumerQueue)){
                    System.out.println(">>>>>>>发送微信消息: "+new String(message.getBody()));
                }
            }
        });
        return simpleMessageListenerContainer;
    }

    /**
     * MessageListenerAdapter
     * 可以把一个没有实现MessageListener、ChannelAwareMessageListener接口的类适配成一个可以处理消息的处理器
     *
     * 默认处理的方法名字为：handleMessage，可以通过setDefaultListenerMethod设置新的消息处理方法
     *
     * MessageListenerAdapter 支持不同的队列交给不同的方法去执行。使用setQueueOrTagToMethodName方法设置，
     * 当根据queue名称没有找到匹配的方法处理的时候，就会交给默认的方法去处理
     *
     */
//    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames("sms","mail","weixin");


        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new MessageHandler());
        messageListenerAdapter.setDefaultListenerMethod("ReceiveMsg");

        /**
         * MessageConverter可以把java对象转换成Message对象，也可以把Message转换成java对象
         *
         * MessageListenerAdapter内部通过MessageConverter把Message转换成java对象，然后找到相应的处理方法，参数为转换成的java对象
         *
         * SimpleMessageConverter处理规则为
         * 1：如果content_type是以text开头，则把消息转换成String
         * 2：如果content_type是application/x-java-serialized-object则把消息反序列化为java对象
         * 否则，把消息转换成byte[]
         *
         */
//        messageListenerAdapter.setMessageConverter(new TextMessageConverter());
        /**
         * Jackson2JsonMessageConverter
         * 如果接收到的消息属性里面没有content_type属性，或者content_type值不包含json，则转换后的结果为byte[]
         *
         */
        Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();

        /**
         * 当消费者有配置映射key的时候，生产者既可以指定java对象全称，又可以是映射的key
         */
        DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
        Map<String,Class<?>> idClassMapping = new HashMap<>();
        idClassMapping.put("order", Order.class);
        idClassMapping.put("user", User.class);
//        idClassMapping.put("user", Person.class);
        javaTypeMapper.setIdClassMapping(idClassMapping);
        jsonConverter.setJavaTypeMapper(javaTypeMapper);

        TextMessageConverter textMessageConverter = new TextMessageConverter();
        ContentTypeDelegatingMessageConverter delegatingMessageConverter = new ContentTypeDelegatingMessageConverter();
        delegatingMessageConverter.addDelegate("text",textMessageConverter);
        delegatingMessageConverter.addDelegate("html/text",textMessageConverter);
        delegatingMessageConverter.addDelegate("xml/text",textMessageConverter);
        delegatingMessageConverter.addDelegate("text/plain",textMessageConverter);

        delegatingMessageConverter.addDelegate("json",jsonConverter);
        delegatingMessageConverter.addDelegate("application/json",jsonConverter);

        FileMessageConverter fileMessageConverter = new FileMessageConverter();
        delegatingMessageConverter.addDelegate("image/jpg",fileMessageConverter);
        delegatingMessageConverter.addDelegate("image/png",fileMessageConverter);
        delegatingMessageConverter.addDelegate("image",fileMessageConverter);
        delegatingMessageConverter.addDelegate("application/pdf",fileMessageConverter);

        messageListenerAdapter.setMessageConverter(delegatingMessageConverter);

//        Map<String,String> queueOrTagToMethodNameMap = new HashMap<>();
//        queueOrTagToMethodNameMap.put("sms","ReceiveSms");
//        queueOrTagToMethodNameMap.put("mail","ReceiveMail");
//        queueOrTagToMethodNameMap.put("weixin","ReceiveWeixin");
//        messageListenerAdapter.setQueueOrTagToMethodName(queueOrTagToMethodNameMap);

        container.setMessageListener(messageListenerAdapter);
        return container;
    }


}
